package cn.com.tt.flink.sql.engine.strategy;

import cn.com.tt.flink.sql.engine.pojo.JobConfig;
import cn.com.tt.flink.sql.engine.pojo.source.JobSource;
import cn.com.tt.flink.sql.engine.pojo.source.SourceField;
import cn.com.tt.flink.sql.engine.pojo.source.SourceProperty;

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;

public class KafkaTableStrategy implements TableStatementStrategy {
    private JobSource source;

    public KafkaTableStrategy(JobSource source) {
        this.source = source;
    }

    @Override
    public void vaildKeyProperties() throws Exception {
        List<SourceProperty> propertyList = source.getPropertyList();
        PropertyValidator validator = new PropertyValidator(propertyList);
        String tableName = source.getTableName();
        validator.vaild("connector", tableName + "缺少connector属性");
        validator.vaild("topic", tableName + "缺少topic属性");
        validator.vaild("properties.bootstrap.servers", tableName + "缺少properties.bootstrap.servers属性");
        validator.vaild("format", tableName + "缺少format属性");
    }

    @Override
    public String createTableStatement() {
        String tableName = source.getTableName();
        StringBuilder sb = new StringBuilder();
        //表头信息
        sb.append("create table ").append(tableName).append("(").append("\n");
        //表字段信息
        sb.append(createTableFields());
        //表属性
        sb.append(") with (").append("\n");
        sb.append(createTableProperties());
        sb.append(")");
        return sb.toString();
    }

    public String createTableFields() {
        List<SourceField> fieldsList = this.source.getFieldsList();
        StringBuilder sb = new StringBuilder();
        fieldsList.forEach(field -> {
            String fieldType = field.getFieldType();

            //自定义类型：proctime(),语法为：列名 as proctime(),添加处理时间列,一般用于temporal join
            if (fieldType.equals("proctime()")) {
                sb.append("`").append(field.getFieldName()).append("`").append(" as ").append(field.getFieldType()).append(",\n");
            } else {
                //其他类型拼接格式： 列名 类型名 精度信息；
                sb.append("`").append(field.getFieldName()).append("`").append(" ").append(field.getFieldType());

                //有则添加精度信息
                if (field.getLength() != 0) {
                    sb.append("(").append(field.getLength());

                    if (field.getDecimals() != 0) {
                        sb.append(",").append(field.getDecimals()).append(")");
                    } else {
                        sb.append(")");
                    }
                }


                sb.append(",\n");
            }
        });

        //拼接primarystr
        Optional<String> primaryFields = fieldsList.stream()
                .filter(SourceField::isPrimaryKey)
                .map(s -> "`" + s.getFieldName() + "`")
                .reduce((a, b) -> a + "," + b);

        primaryFields.ifPresent(p -> sb.append("primary key(").append(p).append(") not enforced,").append("\n"));
        //去掉属性中的最后一个，号
        sb.deleteCharAt(sb.lastIndexOf(","));
        return sb.toString();
    }

    public String createTableProperties() {
        String groupId = source.getJobName();
        String scanStartupMode = JobConfig.getInstance().getScanStartupMode();
        Long scanStartupTimestampMs = JobConfig.getInstance().getScanStartupTimestampMs();
        Map<String, String> proMap = new HashMap<>();
        //默认属性
        proMap.put("properties.group.id", groupId);
        proMap.put("scan.startup.mode", scanStartupMode);

        if (scanStartupMode.equals("timestamp")) {
            proMap.put("scan.startup.timestamp-millis", String.valueOf(scanStartupTimestampMs));
        }
        //加载用户属性
        source.getPropertyList().forEach(p -> proMap.put(p.getPropKey(), p.getPropValue()));
        String reduce = proMap.entrySet().stream()
                .map(e -> "'" + e.getKey() + "' = '" + e.getValue() + "',\n")
                .reduce("", (a, b) -> a + b);

        return reduce.substring(0, reduce.lastIndexOf(","));
    }


}
